package com.atguigu.flink.chapter08;

import com.atguigu.flink.bean.LoginEvent;
import com.atguigu.flink.bean.OrderEvent;
import com.atguigu.flink.util.KKutil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

public class Flink09_Project_Order {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

                env.readTextFile("input/OrderLog.csv")
                        .map(line->{
                            String[] datas = line.split(",");
                            return new OrderEvent(
                                    Long.valueOf(datas[0]),
                                    datas[1],
                                    datas[2],
                                    Long.valueOf(datas[3])
                            );
                        })
                        .assignTimestampsAndWatermarks(WatermarkStrategy
                                .<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((log,es)->log.getEventTime())
                        )
                        .keyBy(OrderEvent::getOrderId)
                        .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
                        .process(new ProcessWindowFunction<OrderEvent, String, Long, TimeWindow>() {

                            private ValueState<OrderEvent> createState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                createState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("createState", OrderEvent.class));
                            }

                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<OrderEvent> elements,
                                                Collector<String> out) throws Exception {

                                List<OrderEvent> list = KKutil.toList(elements);
                                if (list.size()==2) {
                                    out.collect("订单："+key+"正常。。。");
                                }else {
                                    OrderEvent event = list.get(0);
                                    if ("create".equals(event.getEventType())) {

                                        createState.update(event);
                                        //out.collect("订单："+key + "只有创建没有支付，或者超时支付");

                                    }else {
                                        if (createState.value() == null) {

                                            out.collect("订单："+key + "只有支付，没有创建");
                                        }else {
                                            out.collect("订单："+key + "超时支付");

                                        }
                                    }
                                }

                            }
                        }).print();




               try {
                   env.execute();
               } catch (Exception e) {
                   e.printStackTrace();
               }


           }
}
